package com.atguigu.flume.interceptor;

import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Map;

/**
 * kafka-->hdfs
 * 为event的header添加timestamp以便hdfs-sink分区
 */
public class TimeStampInterceptor implements Interceptor {

    JsonParser parser;

    public void initialize() {
        parser=new JsonParser();
    }

    /**
     * 为event的header添加timestamp以便hdfs-sink分区
     * @param event
     * @return
     */
    public Event intercept(Event event) {
        byte[] body = event.getBody();
        JsonElement jsonElement = null;
        try {
            jsonElement = parser.parse(new String(body, "UTF-8"));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        //以下业务不是很严谨,json格式错误就完犊子了
        JsonObject jsonObject = jsonElement.getAsJsonObject();
        String ts = jsonObject.get("ts").getAsString();

        Map<String, String> headers = event.getHeaders();
        headers.put("timestamp",ts);
        return event;
    }


    public List<Event> intercept(List<Event> events) {
        for (Event event : events) {
            intercept(event);
        }
        return events;
    }

    public void close() {

    }

    public static class Builder implements Interceptor.Builder {
        public Interceptor build() {
            return new TimeStampInterceptor();
        }
        /**
         * 配置方法，可以从配置文件中配置我们的Interceptor
         * @param context 配置文件
         */
        public void configure(Context context) {

        }
    }

}
